import time

from kafka import KafkaProducer
from kafka.errors import KafkaError

producer = KafkaProducer(
    #security_protocol="SASL_PLAINTEXT",
    #sasl_mechanism="PLAIN",
    #sasl_plain_username="admin",
    #sasl_plain_password="admin",
    # sasl_plain_password="ffka3&789",
    bootstrap_servers=[
        "172.16.8.29:9092"
        , "172.16.8.29:9093"
                       ])
        # "172.16.8.52:19096", "172.16.8.52:19092", "172.16.8.52:19093", "172.16.8.52:19094", "172.16.8.52:19095"])

# Asynchronous by default
for i in range(10000):
    key = "m" + str(i % 3)
    future = producer.send('kafka_raspberry_pi', value=str.encode('raw_bytes' + str(i)), key=key.encode(), headers="")
    time.sleep(1.3)
    print(i)

# Block for 'synchronous' sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError as e:
    # Decide what to do if produce request failed...
    print(e)
    pass
